#include "../zhelpers.h"
#include "thread.h"
#include "server.h"
#include "proto.msg.pb.h"
#include "Log.h"
#include "vs_opencv.h"
#include "GlobalFeature_API.h"

int gf_extract_thread::thread_func()
{
	if(_tp == NULL){
		LOG_ERROR_VA("pool_work_thread don't belong to a thread_pool");
		return -1;
	}
	void *receiver = zmq_socket(_tp->get_zmq(), ZMQ_REP);
	zmq_connect(receiver, "inproc://workers");
	while(!(_tp->has_stoped())){
		//Wait for request from client
		int size;
		zmq_msg_t recv_msg;
		zmq_msg_init(&recv_msg);

		LOG_DEBUG_VA("start recv...");
		size = zmq_recvmsg(receiver, &recv_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&recv_msg), size);
		proto::front_req proto_msg;
		if(!proto_msg.ParseFromString(proto_buf)){
			LOG_ERROR_VA("ParseFromString fail");
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("recv length: %d, class_id: %d", size, proto_msg.class_id());
		//sleep(1);
		//the work of engine
		char *path_buf = (char *)malloc(proto_msg.image_data().size());
		memcpy(path_buf, proto_msg.image_data().c_str(), proto_msg.image_data().size());
		IplImage *ipl = cv_load_image(path_buf, proto_msg.image_size());
		delete [] path_buf;
		if(ipl == NULL){
			LOG_ERROR_VA("open_cv load error");
			send_error_reply(receiver);
			continue;
		}

		int num_vids = 8;
		int vids[8] = {0};
		unsigned char feature_buf[1000];
		int ret = GLOBALFEATURE_3_2_0::Extraction(ipl, NULL, proto_msg.class_id(), &num_vids, vids, feature_buf);
		if(ret != 0){
			LOG_ERROR_VA("Global Extract failed: %d", ret);
			send_error_reply(receiver);
			continue;
		}
		cv_release_image(ipl);
		database *db = _tp->get_data_base();
		vid_data *vid_datas = new vid_data[num_vids];
		if (db->get_data(proto_msg.class_id(), proto_msg.b2c_flag(), database::gf, num_vids, vids, vid_datas)  <= 0) {
			LOG_ERROR_VA("[gf_match] db get data failed, check data_root_path config");
			send_error_reply(receiver);
			delete [] vid_datas;
			continue;
		}
		GLOBALFEATURE_3_2_0::TDataStruct *datas = new GLOBALFEATURE_3_2_0::TDataStruct[num_vids];
		for (int i = 0; i != num_vids; i++) {
			datas[i].dataLen = vid_datas[i].length;
			datas[i].data = vid_datas[i].data;
		}

		delete [] vid_datas;
		GLOBALFEATURE_3_2_0::TDistributedResult res;
		

		res.nResults = proto_msg.gf_num() > proto_msg.cf_num() ? proto_msg.gf_num(): proto_msg.cf_num();
		res.pImgRes = new GLOBALFEATURE_3_2_0::TDRes[res.nResults];
		res.pClrRes = new GLOBALFEATURE_3_2_0::TDRes[res.nResults];
		res.imageType = 0;
		ret = GLOBALFEATURE_3_2_0::Match(num_vids, datas, feature_buf, proto_msg.merge_same(), &res); 
		if(ret != 0){
			LOG_ERROR_VA("Global Match failed: %d", ret);
			send_error_reply(receiver);
			for (int i = 0; i != num_vids; i++)
				if (datas[i].dataLen)
					delete[] datas[i].data;
			delete [] datas;
			delete [] res.pImgRes;
			delete [] res.pClrRes;
			continue;
		}
		for (int i = 0; i != num_vids; i++)
			if (datas[i].dataLen)
				delete[] datas[i].data;
		delete [] datas;
	
		GLOBALFEATURE_3_2_0::TImageRes *global_res = new GLOBALFEATURE_3_2_0::TImageRes[proto_msg.gf_num()];
		GLOBALFEATURE_3_2_0::TImageRes *color_res = new GLOBALFEATURE_3_2_0::TImageRes[proto_msg.cf_num()];
		int gf_num = proto_msg.gf_num();
		int cf_num = proto_msg.cf_num();
		ret = GLOBALFEATURE_3_2_0::Merge(proto_msg.class_id(), 1, &res, proto_msg.merge_same(), gf_num, global_res, cf_num, color_res);
		if(ret != 0){
			LOG_ERROR_VA("[global] Merge failed: %d", ret);
			send_error_reply(receiver);
			delete [] res.pImgRes;
			delete [] res.pClrRes;
			delete [] global_res;
			delete [] color_res;
			continue;
		}
		delete [] res.pImgRes;
		delete [] res.pClrRes;
		//end of engine
		
		proto::front_resp resp;
		resp.mutable_head()->set_version(420);
		resp.mutable_head()->set_command(420);
		resp.mutable_head()->set_password("xxxxxxxx", 8);
		resp.mutable_head()->set_type(1);
		resp.mutable_head()->set_status(0);
		resp.mutable_head()->set_src(proto_msg.head().src());
		resp.mutable_head()->set_dest(proto_msg.head().dest());
		resp.mutable_head()->set_sn_high(123456789);
		resp.mutable_head()->set_sn_low(123456789);
		resp.mutable_head()->set_length(0);
	
		resp.set_res_num(cf_num + gf_num);
		for(int i = 0; i < cf_num ; i++){
			resp.add_iid(color_res[i].imageID);
			resp.add_gid(color_res[i].groupID);
			resp.add_simi(color_res[i].similarity);
			resp.add_dim(color_res[i].source);
			resp.add_type(color_res[i].imageType);
		}
		for(int i = 0; i < gf_num ; i++){
			resp.add_iid(global_res[i].imageID);
			resp.add_gid(global_res[i].groupID);
			resp.add_simi(global_res[i].similarity);
			resp.add_dim(global_res[i].source);
			resp.add_type(global_res[i].imageType);
		}

		delete [] global_res;
		delete [] color_res;
		string serial_stream;
		if(!resp.SerializePartialToString(&serial_stream)){
			send_error_reply(receiver);
			LOG_ERROR_VA("fail to serial data");
			continue;
		}
		zmq_msg_t send_msg;
		zmq_msg_init_size(&send_msg, serial_stream.size());
		memcpy(zmq_msg_data(&send_msg), serial_stream.c_str(), serial_stream.size());
		size = zmq_sendmsg(receiver, &send_msg, 0);
		if(size == -1){
			send_error_reply(receiver);
			LOG_ERROR_VA("zmq_sendmsg error");
			continue;
		}
	
	//	zmq_msg_t send_msg;
	//	zmq_msg_init_size(&send_msg, 4);
	//	memcpy(zmq_msg_data(&send_msg), &size, 4);
	//	size = zmq_sendmsg(receiver, &send_msg, 0);
		LOG_DEBUG_VA("send over, size: %d", size);
		if(size == -1){
			LOG_ERROR_VA("send error: %d", errno);
			cout << _pid << ": thread send msg ERROR: " << errno << " ; ";
			if(errno == EAGAIN)
				cout << "EAGAIN" << endl;
			else if(errno == ENOTSUP)
				cout << "ENOTSUP" << endl;
			else if(errno == EFSM)
				cout << 3 << endl;
			else if(errno == ETERM)
				cout << 4 << endl;
			else if(errno == ENOTSOCK)
				cout << 5 << endl;
			else if(errno == EINTR)
				cout << 6 << endl;
			else if(errno == EFAULT)
				cout << 7 << endl;
			else 
				cout << "other" << endl;
			/*
			EAGAIN
			ENOTSUP
			EFSM
			ETERM
			ENOTSOCK
			EINTR
			EFAULT
			*/
		}
	}
	zmq_close(receiver);
}

int thread_pool::start()
{
	LOG_INFO_VA("thread_pool start thread num: %d", _thread_num);
	//create all threads
	for(int i = 0; i < _thread_num; i++)
	{
		thread *p = new thread(this);
		_thread_vec.push_back(p);
	}

	//start all the threads
	for (vector<thread *>::iterator it = _thread_vec.begin(); it != _thread_vec.end(); it++) 
	{
		(*it)->start();
	}
}

int thread_pool::stop()
{
	//the diffence beteew this and ~thread_pool()
}

void* thread_pool::get_zmq(){
	return _server->get_zmq();
}

database *thread_pool::get_data_base(){
	return _server->get_data_base();
}

